package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.VisitorStats;
import com.atguigu.gmall.realtime.util.ClickHouseUtil;
import com.atguigu.gmall.realtime.util.DateTimeUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;

/**
 * @author: xu
 * @desc: 访客主题统计
 * 需要启动的服务
 * 程  序：mockLog -> Nginx -> Logger.sh -> Kafka(ZK) -> BaseLogApp -> kafka -> uvApp/ujApp -> Kafka -> VisitorStatsApp -> ClickHouse
 * - 模拟生成日志数据
 * - 交给Nginx进行反向代理
 * - 交给日志处理服务将日志发送到kafka的ods_base_log
 * - BaseLogApp从ods层读取数据，进行分流，将分流的数据发送到kafka的dwd(dwd_page_log)
 * - UniqueVisitApp从dwd_page_log读取数据，将独立访客明细发送到dwm_unique_visit
 * - UserJumpDetailApp从dwd_page_log读取数据，将页面跳出明细发送到dwm_user_jump_detail
 * - VisitorStatsApp
 * - > 从dwd_page_log读取数据，计算pv、持续访问时间、session_count
 * - > 从dwm_unique_visit读取数据，计算uv
 * - > 从dwm_user_jump_detail读取数据，计算页面跳出
 * - > 统一格式，合并
 * - > 分组、开窗、聚合
 * - > 将聚合统计的结果保存到ClickHouse OLAP数据库
 */
public class VisitorStatsApp {
    public static void main(String[] args) throws Exception {
        // TODO 1.基本环境准备
        // 1.1  准备本地测试流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 设置并行度
        env.setParallelism(1);
        // 1.3 设置Checkpoint
        // env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        // env.getCheckpointConfig().setCheckpointTimeout(60000);
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        // env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // env.setStateBackend(new FsStateBackend("hdfs://node1:8020/gmall/checkpoint/VisitorStatsApp"))
        // System.setProperty("HADOOP_USER_NAME", "root");

        // TODO 2.从kafka主题中读取数据
        // 2.1 声明读取的主题名以及消费者组
        String pageViewSourceTopic = "dwd_page_log";
        String uniqueVisitSourceTopic = "dwm_unique_visit";
        String userJumpDetailSourceTopic = "dwm_user_jump_detail";
        String groupId = "visitor_stats_app";
        // 2.2 从dwd_page_log主题中读取日志数据
        DataStreamSource<String> pvJsonStrStream = env.addSource(MyKafkaUtil.getKafkaSource(pageViewSourceTopic, groupId));
        // 2.3 从dwm_unique_visit主题中读取uv数据
        DataStreamSource<String> uvJsonStrStream = env.addSource(MyKafkaUtil.getKafkaSource(uniqueVisitSourceTopic, groupId));
        // 2.4 从dwm_user_jump_detail主题中读取跳出数据
        DataStreamSource<String> userJumpJsonStrStream = env.addSource(MyKafkaUtil.getKafkaSource(userJumpDetailSourceTopic, groupId));
        // 2.5 输出各流中的数据
        // pvJsonStrStream.print("pv>>>>>");
        // uvJsonStrStream.print("uv>>>>>");
        // userJumpJsonStrStream.print("userJump>>>>>");

        // TODO 3.对各个流的数据进行结构的转换，jsonStr->VisitorStats
        // 3.1 转换pv流
        SingleOutputStreamOperator<VisitorStats> pvStatsStream = pvJsonStrStream.map(
                jsonStr -> {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                    JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                    VisitorStats visitorStats = new VisitorStats(
                            "", "",
                            commonJsonObj.getString("vc"),
                            commonJsonObj.getString("ch"),
                            commonJsonObj.getString("ar"),
                            commonJsonObj.getString("is_new"),
                            0L, 1L, 0L, 0L,
                            pageJsonObj.getLong("during_time"),
                            jsonObj.getLong("ts")
                    );
                    // 判断是否为新的会话
                    String lastPageId = pageJsonObj.getString("last_page_id");
                    if (lastPageId == null || lastPageId.length() == 0) {
                        visitorStats.setSv_ct(1L);
                    }
                    return visitorStats;
                }
        );
        // 3.2 转换uv流
        SingleOutputStreamOperator<VisitorStats> uvStatsStream = uvJsonStrStream.map(
                jsonStr -> {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                    return new VisitorStats(
                            "", "",
                            commonJsonObj.getString("vc"),
                            commonJsonObj.getString("ch"),
                            commonJsonObj.getString("ar"),
                            commonJsonObj.getString("is_new"),
                            1L, 0L, 0L, 0L, 0L,
                            jsonObj.getLong("ts")
                    );
                }
        );
        // 3.3 转换跳出流
        SingleOutputStreamOperator<VisitorStats> userJumpStatsStream = userJumpJsonStrStream.map(
                jsonStr -> {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                    return new VisitorStats(
                            "", "",
                            commonJsonObj.getString("vc"),
                            commonJsonObj.getString("ch"),
                            commonJsonObj.getString("ar"),
                            commonJsonObj.getString("is_new"),
                            0L, 0L, 0L, 1L, 0L,
                            jsonObj.getLong("ts")
                    );
                }
        );

        // TODO 4. 将3条流合并到一起，注意：只能合并结构相同的流
        DataStream<VisitorStats> unionStream = pvStatsStream.union(uvStatsStream, userJumpStatsStream);

        // TODO 5.设置Watermark、分组、开窗、聚合
        SingleOutputStreamOperator<VisitorStats> resultStream = unionStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((visitorStats, recordTimestamp) -> visitorStats.getTs())
                )
                // 分组，按照地区、渠道、版本、新老访客维度进行分组，因为我们这里有4个维度，所以将它们封装为一个Tuple4
                .keyBy(new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
                           @Override
                           public Tuple4<String, String, String, String> getKey(VisitorStats visitorStats) throws Exception {
                               return Tuple4.of(visitorStats.getAr(), visitorStats.getCh(), visitorStats.getVc(), visitorStats.getIs_new());
                           }
                       }
                )
                // 开窗 10s的滚动窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 对窗口的数据进行聚合，聚合结束之后，需要补充统计的起止时间
                .reduce(new ReduceFunction<VisitorStats>() {
                            @Override
                            public VisitorStats reduce(VisitorStats stats1, VisitorStats stats2) throws Exception {
                                stats1.setPv_ct(stats1.getPv_ct() + stats2.getPv_ct());
                                stats1.setUv_ct(stats1.getUv_ct() + stats2.getUv_ct());
                                stats1.setSv_ct(stats1.getSv_ct() + stats2.getSv_ct());
                                stats1.setUj_ct(stats1.getUj_ct() + stats2.getUj_ct());
                                stats1.setDur_sum(stats1.getDur_sum() + stats2.getDur_sum());
                                return stats1;
                            }
                        },
                        new ProcessWindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
                            @Override
                            public void process(Tuple4<String, String, String, String> tuple4, Context context, Iterable<VisitorStats> elements, Collector<VisitorStats> out) throws Exception {
                                long start = context.window().getStart();
                                long end =  context.window().getEnd();
                                for (VisitorStats visitorStats : elements) {
                                    System.out.println("============visit??");
                                    // 获取窗口的开始时间
                                    visitorStats.setStt(DateTimeUtil.dateTimeToString(new Date(start)));
                                    // 获取窗口的结束时间
                                    visitorStats.setEdt(DateTimeUtil.dateTimeToString(new Date(end)));
                                    visitorStats.setTs(System.currentTimeMillis());
                                    out.collect(visitorStats);
                                }
                            }
                        }
                );
        resultStream.print(">>>>>");

        // TODO 6.将聚合统计之后的数据写到ClickHouse
        resultStream.addSink(
                ClickHouseUtil.getJdbcSink("insert into visitor_stats values(?,?,?,?,?,?,?,?,?,?,?,?)")
        );

        env.execute(VisitorStatsApp.class.getSimpleName());
    }
}
